package remoting.netty4.client;

import static java.util.concurrent.TimeUnit.MILLISECONDS;

import common.constants.CommonConstants;
import common.exception.RemotingException;
import common.model.Codec2;
import common.model.MagicUrl;
import common.util.ConfigUtils;
import common.util.NetUtils;
import common.util.UrlUtils;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.proxy.Socks5ProxyHandler;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.concurrent.DefaultThreadFactory;
import java.net.InetSocketAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import remoting.AbstractClient;
import remoting.channel.ChannelHandler;
import remoting.netty4.NettyChannel;
import remoting.netty4.codec.NettyCodecAdapter;
import threadpools.ThreadPool;

/**
 * netty client
 *
 * @author xuxueli 2015-11-24 22:25:15
 */
public class NettyClient extends AbstractClient {

  private static final Logger logger = LoggerFactory.getLogger(NettyClient.class);
  /**
   * netty client bootstrap
   */
  private static final NioEventLoopGroup nioEventLoopGroup = new NioEventLoopGroup(
      CommonConstants.DEFAULT_IO_THREADS, new DefaultThreadFactory("NettyClientWorker", true));

  private static final String SOCKS_PROXY_HOST = "socksProxyHost";

  private static final String SOCKS_PROXY_PORT = "socksProxyPort";

  private static final String DEFAULT_SOCKS_PROXY_PORT = "1080";

  private Bootstrap bootstrap;

  private volatile Channel channel;

  /**
   * The constructor of NettyClient. It wil init and start netty.
   */
  public NettyClient(final MagicUrl url, final ChannelHandler handler, ThreadPool threadPool,
      Codec2 codec) throws RemotingException {
    // you can customize name and type of client thread pool by THREAD_NAME_KEY and THREADPOOL_KEY in CommonConstants.
    // the handler will be warped: MultiMessageHandler->HeartbeatHandler->handler
    super(url, wrapChannelHandler(url, handler, threadPool), codec);
  }

  /**
   * Init bootstrap
   *
   * @throws Throwable
   */
  @Override
  protected void doOpen() throws Throwable {
    final NettyClientHandler nettyClientHandler = new NettyClientHandler(getUrl(), this, getCodec());
    bootstrap = new Bootstrap();
    bootstrap.group(nioEventLoopGroup)
        .option(ChannelOption.SO_KEEPALIVE, true)
        .option(ChannelOption.TCP_NODELAY, true)
        .option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
        //.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, getTimeout())
        .channel(NioSocketChannel.class);

    bootstrap.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, Math.max(3000, getConnectTimeout()));
    bootstrap.handler(new ChannelInitializer() {

      @Override
      protected void initChannel(Channel ch) throws Exception {
        int heartbeatInterval = UrlUtils.getHeartbeat(getUrl());
        NettyCodecAdapter adapter = new NettyCodecAdapter(getCodec(), getUrl(), NettyClient.this);
        ch.pipeline()//.addLast("logging",new LoggingHandler(LogLevel.INFO))//for debug
            .addLast("decoder", adapter.getDecoder())
            .addLast("encoder", adapter.getEncoder())
            .addLast("client-idle-handler",
                new IdleStateHandler(heartbeatInterval, 0, 0, MILLISECONDS))
            .addLast("handler", nettyClientHandler);
        String socksProxyHost = ConfigUtils.getProperty(SOCKS_PROXY_HOST);
        if (socksProxyHost != null) {
          int socksProxyPort = Integer
              .parseInt(ConfigUtils.getProperty(SOCKS_PROXY_PORT, DEFAULT_SOCKS_PROXY_PORT));
          Socks5ProxyHandler socks5ProxyHandler = new Socks5ProxyHandler(
              new InetSocketAddress(socksProxyHost, socksProxyPort));
          ch.pipeline().addFirst(socks5ProxyHandler);
        }
      }
    });
  }

  @Override
  protected void doConnect() throws Throwable {
    long start = System.currentTimeMillis();
    ChannelFuture future = bootstrap.connect(getConnectAddress());
    try {
      boolean ret = future.awaitUninterruptibly(getConnectTimeout(), MILLISECONDS);

      if (ret && future.isSuccess()) {
        Channel newChannel = future.channel();
        try {
          // Close old channel
          // copy reference
          Channel oldChannel = NettyClient.this.channel;
          if (oldChannel != null) {
            try {
              if (logger.isInfoEnabled()) {
                logger.info(
                    "Close old netty channel " + oldChannel + " on create new netty channel "
                        + newChannel);
              }
              oldChannel.close();
            } finally {
              NettyChannel.removeChannelIfDisconnected(oldChannel);
            }
          }
        } finally {
          if (NettyClient.this.isClosed()) {
            try {
              if (logger.isInfoEnabled()) {
                logger
                    .info("Close new netty channel " + newChannel + ", because the client closed.");
              }
              newChannel.close();
            } finally {
              NettyClient.this.channel = null;
              NettyChannel.removeChannelIfDisconnected(newChannel);
            }
          } else {
            NettyClient.this.channel = newChannel;
          }
        }
      } else if (future.cause() != null) {
        throw new RemotingException(this,
            "client(url: " + getUrl() + ") failed to connect to server "
                + getRemoteAddress() + ", error message is:" + future.cause().getMessage(),
            future.cause());
      } else {
        throw new RemotingException(this,
            "client(url: " + getUrl() + ") failed to connect to server "
                + getRemoteAddress() + " client-side timeout "
                + getConnectTimeout() + "ms (elapsed: " + (System.currentTimeMillis() - start)
                + "ms) from netty client "
                + NetUtils.getLocalHost() + " using magic version " + CommonConstants.VERSION);
      }
    } finally {
      // just add new valid channel to NettyChannel's cache
      if (!isConnected()) {
        //future.cancel(true);
      }
    }
  }

  @Override
  protected void doDisConnect() throws Throwable {
    try {
      NettyChannel.removeChannelIfDisconnected(channel);
    } catch (Throwable t) {
      logger.warn(t.getMessage());
    }
  }

  @Override
  protected void doClose() throws Throwable {
  }

  @Override
  protected remoting.channel.Channel getChannel() {
    Channel c = channel;
    if (c == null || !c.isActive()) {
      return null;
    }
    return NettyChannel.getOrAddChannel(c, getUrl(), this, getCodec());
  }

  @Override
  public boolean canHandleIdle() {
    return true;
  }
}
